package com.heima.article.kafkastream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constant.GlobalConstant;
import com.heima.model.article.dots.UpdateArticleMsg;
import lombok.extern.slf4j.Slf4j;

import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class HotArticleStream {
    @Bean
    public KStream<String,String> KStream(StreamsBuilder streamsBuilder){
        KStream<String, String> stream = streamsBuilder.stream(GlobalConstant.HOT_ARTICLE_SCORE_STREAM_TOPIC);
        stream.map((key,value)->{
            UpdateArticleMsg updateArticleMsg = JSON.parseObject(value, UpdateArticleMsg.class);
            return new KeyValue<>(updateArticleMsg.getArticleId().toString(),value);
        }).groupBy((key,value)->key)
                .windowedBy(TimeWindows.of(Duration.ofSeconds(GlobalConstant.KAFKA_STREAM_INTERVAL)))
                .aggregate(new Initializer<String>() {
                    @Override
                    public String apply() {
                        return JSON.toJSONString(UpdateArticleMsg.getInit());
                    }
                }, new Aggregator<String, String, String>() {

                    @Override
                    public String apply(String key, String newValue, String oldValue) {
                        log.info(" kafka stream:new->{},old->{}",newValue,oldValue);
                        UpdateArticleMsg newMsg=JSON.parseObject(newValue,UpdateArticleMsg.class);
                        UpdateArticleMsg oldMsg=JSON.parseObject(oldValue,UpdateArticleMsg.class);
                        newMsg.add(oldMsg);
                        return JSON.toJSONString(newMsg);
                    }
                },Materialized.as(GlobalConstant.KAFKA_STREAM_MATERIALIZED))
                .toStream()
                .map(new KeyValueMapper<Windowed<String>, String, KeyValue<?, ?>>() {
                    @Override
                    public KeyValue<?, ?> apply(Windowed<String> windowed, String msgJson) {
                        log.info("total->{}",msgJson);
                        return new KeyValue<>(windowed.key(),msgJson);
                    }
                })
                .to(GlobalConstant.KAFKA_STREAM_BATCH_TOPIC);
        return stream;
    }
}
